package datastream.api;

import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.environment.LocalStreamEnvironment;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class ExecutionEnvironment {
    public static void main(String[] args) {
        ParameterTool parameterTool = ParameterTool.fromArgs(args);
        Configuration configuration = Configuration.fromMap(parameterTool.toMap());

        // 1 生产环境最常用的方式
        // Flink 自动判断当前作业的执行环境
        // --- 本地环境（IDE)：返回本地执行环境运行 Flink 作业
        // --- 集群环境：返回集群的执行环境（需搭建集群环境，构建 Flink 作业 JAR 包，并通过 Flink 命令行向集群提交作业）
        StreamExecutionEnvironment env1 = StreamExecutionEnvironment
                .getExecutionEnvironment();

        // 2 创建本地执行环境
        LocalStreamEnvironment env2 = StreamExecutionEnvironment
                .createLocalEnvironment();

        // 2 创建包含 Flink Web UI 的本地执行环境
        // 通过 localhost:8081 查看 Flink Web UI
        StreamExecutionEnvironment env3 = StreamExecutionEnvironment
                .createLocalEnvironmentWithWebUI(configuration);

        // 2 通过 getExecutionEnvironment 创建带 Flink Web UI 的执行环境
        configuration.setInteger(RestOptions.PORT, RestOptions.PORT.defaultValue());
        StreamExecutionEnvironment env4 = StreamExecutionEnvironment.getExecutionEnvironment(configuration);


        // 3 向集群环境提交 Flink 作业
        // 需指定 JobManager 主机、端口、需执行的 JAR 包路径
        // 单并行，除非通过 setParallelism 显式指定
        StreamExecutionEnvironment env5 = StreamExecutionEnvironment
                .createRemoteEnvironment("localhost", 8000, "");
    }
}
